package cn.com.netedge.iot.common.utils;

import cn.com.netedge.iot.common.constants.Globals;
import cn.com.netedge.iot.common.constants.MqConstants;
import cn.com.netedge.iot.common.mq.MqMessageListener;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.PostConstruct;
import java.util.List;

@RefreshScope
public abstract class BaseJmsMsgUtil {
    private static final Logger logger = LoggerFactory.getLogger(BaseJmsMsgUtil.class);
    private static final int HEALTH_CHECK_INTERVAL = 10 * 1000;
    @Value("${rabbitmq.host}")
    private String host;
    @Value("${rabbitmq.port}")
    private int port;
    @Value("${rabbitmq.username}")
    private String username;
    @Value("${rabbitmq.password}")
    private String password;
    @Value("${rabbitmq.exchange}")
    private String exchange;

    private static String appName;
    private List<String> quequeList;

    private static RabbitTemplate rabbitTemplate;
    private CachingConnectionFactory connectionFactory;

    private static volatile boolean mqHealth = false;

    @PostConstruct
    protected void initConnection() {
        connectionFactory = buildFactory();
        rabbitTemplate = new RabbitTemplate(connectionFactory);
    }

    private CachingConnectionFactory buildFactory() {
        logger.debug("build connectionFactory, host:{}, port:{}, username:{}, password:{}, exchange:{}", host, port, username, password, exchange);
        connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    public abstract void subscribeQueues();

    public static boolean sendMessage(String queueName, String message) {
        if (StringUtil.isEmpty(message)) {
            return false;
        }
        if (!mqHealth) {
            if (needLocalQueue()) {
                ChronicleQueueUtil.enqueue(queueName, message);
                return false;
            }
        }
        try {
            rabbitTemplate.convertAndSend(queueName, message);
            logger.debug("sent message successfully:{} -> {}", queueName, message);
            return true;
        } catch (Exception e) {
            logger.debug("send message error.", e);
            retrySendMessage(queueName, message);
        }
        return false;
    }

    @Scheduled(fixedDelay = HEALTH_CHECK_INTERVAL)
    public void isKafkaAvailable() {
        try {
            rabbitTemplate.convertAndSend("hello", "hello world!");
            mqHealth = true;
        } catch (Exception e) {
            logger.debug("send message error.", e);
            mqHealth = false;
        }
    }

    private static void retrySendMessage(String queueName, String message) {
        if (needLocalQueue()) {
            ChronicleQueueUtil.enqueue(queueName, message);
            return;
        }
        try {
            Thread.sleep(30000);
        } catch (Exception e) {
            logger.error("sleep error.", e);
        }
        sendMessage(queueName, message);
    }

    protected void subscribeQueue(List<String> quequeList) {
        if (quequeList == null || quequeList.isEmpty()) {
            return;
        }
        this.quequeList = quequeList;
        for (String queueName : quequeList) {
            subscribeQueue(queueName);
        }
    }

    private static boolean needLocalQueue() {
        if (Globals.APP_FRONTEND_COMM.equals(appName) || Globals.APP_FRONTEND_BIZ.equals(appName) || Globals.APP_FRONTEND_DATA_SAVE.equals(appName)) {
            return true;
        }
        return false;
    }

    public void subscribeQueue(String queueName) {
        if (connectionFactory == null) {
            connectionFactory = buildFactory();
        }
        subscribeQueue(queueName, connectionFactory);
    }

    public abstract MqMessageListener getMessageListener(String queueName);

    public void subscribeQueue(String queueName, ConnectionFactory connectionFactory) {
        if (MqConstants.containsMqMessageListener(queueName)) {
            logger.trace("queueName:{} already exists", queueName);
            return;
        }
        //创建队列
        try (Connection connection = connectionFactory.createConnection();
             Channel channel = connection.createChannel(true)) {

            // 声明队列（如果队列已存在，调用将忽略）
            channel.queueDeclare(queueName, true, false, false, null);
            logger.info("Queue:{} declared successfully or already exists.",queueName);
        } catch (Exception e) {
           logger.info("create rabbitmq queue error.", e);
        }
        logger.debug("queueName:{} subscribe", queueName);
        MqMessageListener listener = getMessageListener(queueName);
        if (listener == null) {
            return;
        }
        listener.setQueueName(queueName);
        listener.setConnectionFactory(connectionFactory);
        listener.start();
    }

    public static void setAppName(String name) {
        appName = name;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public void setExchange(String exchange) {
        this.exchange = exchange;
    }
}
